package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TradeSkuOrderBean;
import com.atguigu.bean.TradeSkuOrderBean9;
import com.atguigu.func.DimAsyncJoinFunction;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.KafkaUtil;
import com.atguigu.utils.MyClickhouseUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.swing.*;
import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/*
10.9 交易域商品粒度下单各窗口汇总表
10.9.1 主要任务(维度关联，旁路缓存+异步IO)
从 Kafka 订单明细主题读取数据，过滤null数据，按照唯一键去重，分组开窗聚合，统计各维度各窗口的原始金额、活动减免金额、优惠券减免金额和订单金额，补全维度信息，将数据写入 ClickHouse 交易域SKU粒度下单各窗口汇总表。

 */
//todo 1.创建环境
//todo 2.读取kafka dwd订单明细事实表数据
//todo 3.过滤null值（订单明细join订单 left join优惠卷明细left join活动明细look up join字典表），并转化为jsonObject
//todo 4.提取时间时间戳并生成watermark
//todo 5.按照订单明细id分组，去重由上游left join产生的重复数据（左null、左右）
//todo 6.按照sku_id分组，求四个度量值，并转化为javabean
//todo 开窗、聚合（先聚合再关联维表）
//todo 7.关联维表，获取维表信息（异步IO+旁路缓存）
//todo 8.将数据写到clickhouse
//todo 9.启动任务

public class Dws09TradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //todo 1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 2.读取kafka dwd订单明细事实表数据
        DataStreamSource<String> kafkaDS = env.addSource(KafkaUtil.getFlinkKafkaConsumer("dwd_trade_order_detail", "skuorder"));

        //todo 3.过滤null值（订单明细join订单 left join优惠卷明细left join活动明细look up join字典表），并转化为jsonObject
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                }
            }
        });

        //todo 4.提取时间时间戳并生成watermark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWMDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("create_time");
                            }
                        })
        );

        //todo 5.按照订单明细id分组，去重由上游left join产生的重复数据（左null、左右）
        KeyedStream<JSONObject, String> keyByIdStream = jsonObjWithWMDS.keyBy(s -> s.getString("id"));
        SingleOutputStreamOperator<JSONObject> jsonObjWithOneIdDS = keyByIdStream.flatMap(new RichFlatMapFunction<JSONObject, JSONObject>() {
            //定义状态
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //初始化状态
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("id-state", String.class);
                StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.seconds(10))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        .build();
                valueStateDescriptor.enableTimeToLive(ttlConfig);

                valueState = getRuntimeContext().getState(valueStateDescriptor);

            }

            @Override
            public void flatMap(JSONObject value, Collector<JSONObject> out) throws Exception {
                //获取状态
                String stateValue = valueState.value();

                //过滤掉重复的订单明细id，由于度量值是左表的，所以不管是左null、左右谁先来，留最先来的那个就可以了
                String id = value.getString("id");
                if (stateValue == null) {
                    out.collect(value);
                    valueState.update("1");//随便存什么值都可以
                }
            }
        });


        //todo 6.按照sku_id分组，求四个度量值，并转化为javabean
        KeyedStream<JSONObject, String> keyBySkuIdStream = jsonObjWithOneIdDS.keyBy(s -> s.getString("sku_id"));
        SingleOutputStreamOperator<TradeSkuOrderBean9> tradeSkuOrderBeanDS = keyBySkuIdStream.flatMap(new RichFlatMapFunction<JSONObject, TradeSkuOrderBean9>() {
            @Override
            public void flatMap(JSONObject value, Collector<TradeSkuOrderBean9> out) throws Exception {
                String sku_id = value.getString("sku_id");
                String sku_name = value.getString("sku_name");
                //获取四个度量值
                BigDecimal split_coupon_amount = value.getBigDecimal("split_coupon_amount");
                BigDecimal split_activity_amount = value.getBigDecimal("split_activity_amount");
                BigDecimal split_total_amount = value.getBigDecimal("split_total_amount");
                BigDecimal split_original_amount = value.getBigDecimal("sku_num").multiply(value.getBigDecimal("order_price"));

                value.getBigDecimal("");
                out.collect(TradeSkuOrderBean9.builder()
                        .skuId(sku_id)
                        .skuName(sku_name)
                        .couponAmount(split_coupon_amount)
                        .originalAmount(split_original_amount)
                        .orderAmount(split_total_amount)
                        .activityAmount(split_activity_amount)
                        .build()
                );


            }
        });

        //todo 开窗、聚合（先聚合再关联维表）
        SingleOutputStreamOperator<TradeSkuOrderBean9> reducedDS = tradeSkuOrderBeanDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TradeSkuOrderBean9>() {
                    @Override
                    public TradeSkuOrderBean9 reduce(TradeSkuOrderBean9 value1, TradeSkuOrderBean9 value2) throws Exception {
                        value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                        value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        return value1;

                    }
                }, new AllWindowFunction<TradeSkuOrderBean9, TradeSkuOrderBean9, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TradeSkuOrderBean9> values, Collector<TradeSkuOrderBean9> out) throws Exception {
                        TradeSkuOrderBean9 next = values.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setTs(System.currentTimeMillis());

                    }
                });

        //todo 7.关联维表，获取维表信息（异步IO+旁路缓存）
        //关联sku（获取spu_id和tm_id)
        SingleOutputStreamOperator<TradeSkuOrderBean9> reducesWithSkuDS = AsyncDataStream.unorderedWait(reducedDS, new DimAsyncJoinFunction<TradeSkuOrderBean9>("DIM_SKU_INFO") {
            @Override
            public String getKey(TradeSkuOrderBean9 input) {
                return input.getSkuId();
            }

            @Override
            public void join(TradeSkuOrderBean9 input, JSONObject dimInfo) {
                input.setSkuName(dimInfo.getString("SKU_NAME"));

                input.setSpuId(dimInfo.getString("SPU_ID"));
                input.setTrademarkId(dimInfo.getString("TM_ID"));
                input.setCategory3Id(dimInfo.getString("CATEGORY3_ID"));

            }
        }, 60, TimeUnit.SECONDS);

        //关联spu(获取spu_name)
        SingleOutputStreamOperator<TradeSkuOrderBean9> reducedWithSpuDS = AsyncDataStream.unorderedWait(reducesWithSkuDS, new DimAsyncJoinFunction<TradeSkuOrderBean9>("DIM_SPU_INFO") {
            @Override
            public String getKey(TradeSkuOrderBean9 input) {
                return input.getSpuId();
            }

            @Override
            public void join(TradeSkuOrderBean9 input, JSONObject dimInfo) {
                input.setSpuName(dimInfo.getString("SPU_NAME"));

            }
        }, 60, TimeUnit.SECONDS);

        //关联trademark(获取tm_name)
        SingleOutputStreamOperator<TradeSkuOrderBean9> reducedWithTMDS = AsyncDataStream.unorderedWait(reducedWithSpuDS, new DimAsyncJoinFunction<TradeSkuOrderBean9>("DIM_BASE_TRADEMARK") {
            @Override
            public String getKey(TradeSkuOrderBean9 input) {
                return input.getTrademarkId();
            }

            @Override
            public void join(TradeSkuOrderBean9 input, JSONObject dimInfo) {
                input.setTrademarkName(dimInfo.getString("TM_NAME"));

            }
        }, 60, TimeUnit.SECONDS);

        //关联category3(获取c3_name,c2_id) c3\c2\c1顺序不能变
        SingleOutputStreamOperator<TradeSkuOrderBean9> reducedWithC3DS = AsyncDataStream.unorderedWait(reducedWithTMDS, new DimAsyncJoinFunction<TradeSkuOrderBean9>("DIM_BASE_CATEGOTY3") {
            @Override
            public String getKey(TradeSkuOrderBean9 input) {
                return input.getCategory3Id();
            }

            @Override
            public void join(TradeSkuOrderBean9 input, JSONObject dimInfo) {
                input.setCategory2Id(dimInfo.getString("CATEGORY2_ID"));

                input.setCategory3Name(dimInfo.getString("NAME"));

            }
        }, 60, TimeUnit.SECONDS);

        //关联category2(获取c2_name,c1_id)
        SingleOutputStreamOperator<TradeSkuOrderBean9> reducdWithC2DS = AsyncDataStream.unorderedWait(reducedWithC3DS, new DimAsyncJoinFunction<TradeSkuOrderBean9>("DIM_BASE_CATEGORY2") {
            @Override
            public String getKey(TradeSkuOrderBean9 input) {
                return input.getCategory2Id();
            }

            @Override
            public void join(TradeSkuOrderBean9 input, JSONObject dimInfo) {
                input.setCategory2Name(dimInfo.getString("NAME"));
                input.setCategory1Id(dimInfo.getString("CATEGORY1_ID"));

            }
        }, 60, TimeUnit.SECONDS);



        //关联category1(获取c1_name)
        SingleOutputStreamOperator<TradeSkuOrderBean9> resultDS = AsyncDataStream.unorderedWait(reducdWithC2DS, new DimAsyncJoinFunction<TradeSkuOrderBean9>("DIM_BASE_CATEGORY1") {
            @Override
            public String getKey(TradeSkuOrderBean9 input) {
                return input.getCategory1Id();
            }

            @Override
            public void join(TradeSkuOrderBean9 input, JSONObject dimInfo) {

                input.setCategory1Name(dimInfo.getString("NAME"));

            }
        }, 60, TimeUnit.SECONDS);
        resultDS.print("即将写入clickhouse的数据：");


        //todo 8.将数据写到clickhouse
        resultDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_trade_sku_order_window values(?,?,?,?,?,?,?,?,?,?,?,)"));
        //todo 9.启动任务
        env.execute("Dws09TradeSkuOrderWindow");


    }
}
